package com.spx.chapter07;

import com.spx.chapter05.pojo.Event;
import com.spx.util.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;
import java.util.*;

/**
 * create by undeRdoG on  2022-05-03  16:03
 * 凡心所向，素履以往，生如逆旅，一苇以航。
 */
public class TopNProcessAllWindowFunction {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                );


        stream.print("input");

        // 不分组，直接开窗
        stream.map(data -> data.url)
                .windowAll(SlidingEventTimeWindows.of(Time.seconds(10L), Time.seconds(5L)))
                .aggregate(new UrlHashMapCountAgg(), new UrlAllWindowResult())
                .print("Top2");


        env.execute();
    }



    public static class UrlHashMapCountAgg implements AggregateFunction<String, HashMap<String,Long>, List<Tuple2<String,Long>>>{

        @Override
        public HashMap<String, Long> createAccumulator() {
            return new HashMap<>();
        }

        @Override
        public HashMap<String, Long> add(String value, HashMap<String, Long> accumulator) {
            Long val = accumulator.get(value);
            accumulator.put(value, val == null ? 1L : val + 1);
            return accumulator;
        }

        @Override
        public List<Tuple2<String, Long>> getResult(HashMap<String, Long> accumulator) {
            List<Tuple2<String,Long>> res = new ArrayList<>();
            for (Map.Entry<String, Long> entry : accumulator.entrySet()) {
                res.add(Tuple2.of(entry.getKey(), entry.getValue()));
            }

            res.sort(new Comparator<Tuple2<String, Long>>() {
                @Override
                public int compare(Tuple2<String, Long> o1, Tuple2<String, Long> o2) {
                    // 降序排列
                    if (o1.f1.equals(o2.f1)){
                        return 0;
                    }else if (o1.f1 < o2.f1){
                        return 1;
                    }else {
                        return -1;
                    }
                }
            });

            return res;
        }

        @Override
        public HashMap<String, Long> merge(HashMap<String, Long> a, HashMap<String, Long> b) {
            return null;
        }
    }



    public static class UrlAllWindowResult extends ProcessAllWindowFunction<List<Tuple2<String,Long>>,String, TimeWindow>{

        @Override
        public void process(Context context, Iterable<List<Tuple2<String, Long>>> elements, Collector<String> out) throws Exception {
            StringBuilder builder = new StringBuilder();

            builder.append("--------------------------------------------").append("\n");
            builder.append("current window finish time: " + new Timestamp(context.window().getEnd())).append("\n");
            List<Tuple2<String, Long>> list = elements.iterator().next();
            for (int i = 0; i < 2; i++){
                Tuple2<String, Long> tuple2 = list.get(i);
                builder.append("url is " + tuple2.f0 + " and count is " + tuple2.f1).append("\n");
            }

            out.collect(builder.toString());
        }
    }

}
